package com.zs.service.rocketmq.consumer;

import com.zs.framework.cache.RedisCache;
import com.zs.common.annotation.ConsumerProvider;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * Copyright 2022 ZS Network Co., Ltd. All rights reserved.
 * 本文件仅限于正数网络技术有限公司内部传阅，禁止外泄以及用于其他的商业目的。
 * Created by 张晓珂 on 2022/11/7.
 */
@Component
@ConsumerProvider(groupName = "demo-producer-group",topic = "demo-topic",tag = "demo-topic-tag")
@Slf4j
public class DemoRocketMqConsumer implements MessageListenerConcurrently, InitializingBean {

    @Autowired
    private RedisCache redisCache;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        if (CollectionUtils.isEmpty(list)) {
            log.debug("MQ接收消息为空，直接返回成功");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        for (MessageExt messageExt : list) {
            String body = null;
            body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
            log.info("rocketmq msg:{}",body);
            // todo: 填写具体的业务处理逻辑……
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("test");
    }
}